package com.lifq.flink.table;

import com.ververica.cdc.connectors.tidb.TDBSourceOptions;
import com.ververica.cdc.connectors.tidb.TiDBSource;
import com.ververica.cdc.connectors.tidb.TiKVChangeEventDeserializationSchema;
import com.ververica.cdc.connectors.tidb.TiKVSnapshotEventDeserializationSchema;
import com.ververica.cdc.connectors.tidb.table.StartupOptions;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.tikv.kvproto.Cdcpb;
import org.tikv.kvproto.Kvrpcpb;

import java.util.HashMap;

/**
 * @Author: LiFuqiang
 * @Date: 2023/8/15 9:40
 * @Version 1.0
 * @description
 */
public class TiDBSourceExample {

    public static void main(String[] args) throws Exception {
        tt2();
    }

    public static void tt2(){
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(srcTableSql());
        tableEnv.executeSql(destTableSql());
        tableEnv.createStatementSet().addInsertSql(insertSql()).execute();
    }

    public static String srcTableSql(){
        return "CREATE TABLE tbl_sim_day_usage_detail_new_src_flink (\n" +
                "     `id` BIGINT,\n" +
                "  `serial` STRING,\n" +
                "  `msisdn` STRING,\n" +
                "  `eid` STRING,\n" +
                "  `lifecycle_serial` STRING,\n" +
                "  `ec_cust_serial` STRING,\n" +
                "  `biz_type` STRING,\n" +
                "  `apn_name` STRING,\n" +
                "  `pcc_code` STRING,\n" +
                "  `offering_type_detail` STRING,\n" +
                "  `traffic_usage` BIGINT,\n" +
                "  `total_usage` BIGINT,\n" +
                "  `day` STRING,\n" +
                "  `month` STRING,\n" +
                "  `create_time` TIMESTAMP(0),\n" +
                "  `create_by` STRING,\n" +
                "  `modify_time` TIMESTAMP(0),\n" +
                "  `modify_by` STRING,\n" +
                "  `is_delete` INT,\n" +
                "  `lifecycle_node_serial` STRING,\n" +
                "  PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (\n" +
                "     'connector' = 'tidb-cdc',\n" +
                "     'tikv.grpc.timeout_in_ms' = '20000',\n" +
                "     'pd-addresses' = '10.71.37.24:2379,10.71.37.23:2379,10.71.37.25:2379',\n" +
                "     'database-name' = 'traffic_pp',\n" +
                "     'scan.startup.mode' = 'initial',\n" +
                "     'table-name' = 'tbl_sim_day_usage_detail_new'\n" +
                ")";
    }

    public static String destTableSql(){
        return "CREATE TABLE tbl_sim_day_usage_detail_new_dst_flink(\n" +
                "  `id` BIGINT,\n" +
                "  `create_time` TIMESTAMP(0),\n" +
                "  `serial` STRING,\n" +
                "  `msisdn` STRING,\n" +
                "  `eid` STRING,\n" +
                "  `lifecycle_serial` STRING,\n" +
                "  `ec_cust_serial` STRING,\n" +
                "  `biz_type` STRING,\n" +
                "  `apn_name` STRING,\n" +
                "  `pcc_code` STRING,\n" +
                "  `offering_type_detail` STRING,\n" +
                "  `traffic_usage` BIGINT,\n" +
                "  `total_usage` BIGINT,\n" +
                "  `day` STRING,\n" +
                "  `month` STRING,\n" +
                "  `create_by` STRING,\n" +
                "  `modify_time` TIMESTAMP(0),\n" +
                "  `modify_by` STRING,\n" +
                "  `is_delete` INT,\n" +
                "  `lifecycle_node_serial` STRING,\n" +
                "    PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "   'connector' = 'doris',\n" +
                "   'fenodes' = '10.0.61.72:28030,10.0.61.71:28030',\n" +
                "   'table.identifier' = 'example_db.tbl_sim_day_usage_detail_lifq_3',\n" +
                "   'username' = 'root',\n" +
                "   'password' = 'YHNa_Gu0&',\n" +
                "   'sink.label-prefix' = '"+System.currentTimeMillis()+"',\n" +
                "   'sink.properties.format' = 'json',\n" +
                "   'sink.properties.read_json_by_line' = 'true',\n" +
                "   'sink.properties.strict_mode' = 'false',\n" +
                "   'sink.enable-2pc' = 'false',\n" +
                "   'sink.buffer-size' = '4194304',\n" +
                "   'sink.batch.interval' = '10'\n" +
                ")";
    }

    public static String insertSql(){
        return "insert into `tbl_sim_day_usage_detail_new_dst_flink`\n" +
                "(`id`,`create_time`,`serial`,`msisdn`,`eid`,`lifecycle_serial`,`ec_cust_serial`,`biz_type`,`apn_name`,`pcc_code`,`offering_type_detail`,`traffic_usage`,`total_usage`,`day`,`month`,`create_by`,`modify_time`,`modify_by`,`is_delete`,`lifecycle_node_serial` )\n" +
                "select\n" +
                "`id`,`create_time`,`serial`,`msisdn`,`eid`,`lifecycle_serial`,`ec_cust_serial`,`biz_type`,`apn_name`,`pcc_code`,`offering_type_detail`,`traffic_usage`,`total_usage`,`day`,`month`,`create_by`,`modify_time`,`modify_by`,`is_delete`,`lifecycle_node_serial`\n" +
                "From `tbl_sim_day_usage_detail_new_src_flink`";
    }

    public static void tt1() throws Exception {
        SourceFunction<String> tidbSource =
                TiDBSource.<String>builder()
                        .database("mydb") // set captured database
                        .tableName("products") // set captured table
                        .tiConf(
                                TDBSourceOptions.getTiConfiguration(
                                        "localhost:2399", new HashMap<>()))
                        .snapshotEventDeserializer(
                                new TiKVSnapshotEventDeserializationSchema<String>() {
                                    @Override
                                    public void deserialize(
                                            Kvrpcpb.KvPair record, Collector<String> out)
                                            throws Exception {
                                        out.collect(record.toString());
                                    }

                                    @Override
                                    public TypeInformation<String> getProducedType() {
                                        return BasicTypeInfo.STRING_TYPE_INFO;
                                    }
                                })
                        .changeEventDeserializer(
                                new TiKVChangeEventDeserializationSchema<String>() {
                                    @Override
                                    public void deserialize(
                                            Cdcpb.Event.Row record, Collector<String> out)
                                            throws Exception {
                                        out.collect(record.toString());
                                    }

                                    @Override
                                    public TypeInformation<String> getProducedType() {
                                        return BasicTypeInfo.STRING_TYPE_INFO;
                                    }
                                })
                        .startupOptions(StartupOptions.initial())
                        .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        env.enableCheckpointing(3000);
//        DataStreamSource<String> source =  env.addSource(tidbSource);
//        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create();


        env.execute("Print TiDB Snapshot + Binlog");
    }
}
